#pragma once
#include <muduo/net/EventLoopThread.h>
#include <muduo/base/CountDownLatch.h>
#include <muduo/net/TcpConnection.h>
#include <muduo/net/TcpServer.h>
#include <muduo/net/EventLoop.h>
#include <muduo/net/TcpClient.h>
#include <muduo/net/Buffer.h>
#include <unordered_map>
#include <iostream>
#include <string>
#include <mutex>
#include "abstract.hpp"
#include "fields.hpp"
#include "detail.hpp"
#include "message.hpp"

/**网络部分
 *
 */

namespace Json_rpc
{
    class MuduoBuffer : public BaseBuffer
    {
    public:
        using ptr = std::shared_ptr<MuduoBuffer>;
        MuduoBuffer(muduo::net::Buffer *buf) : _buf(buf)
        {
        }
        // 获取缓冲区已有数据
        virtual size_t readableSize() override
        {
            return _buf->readableBytes();
        }
        // 尝试取出4字节数据 不删除数据
        virtual int32_t peeKInt32() override
        {
            // muduo库是从缓冲区取出4字节整形，会进行网络字节序转换为主机字节序
            return _buf->peekInt32();
        }
        // 从缓冲区删除4字节数据
        virtual void retrieveInt32() override
        {
            _buf->retrieveInt32();
        }
        // 尝试取出4字节数据并且删除数据
        virtual int32_t readInt32() override
        {
            return _buf->readInt32();
        }
        // 从缓冲区取出指定长度数据
        virtual std::string retrieveAsString(size_t len) override
        {
            return _buf->retrieveAsString(len);
        }

    private:
        muduo::net::Buffer *_buf;
    };

    class BufferFactory
    {
    public:
        template <typename... Args>
        static BaseBuffer::ptr create(Args &&...args)
        {
            return std::make_shared<MuduoBuffer>(std::forward<Args>(args)...);
        }
    };

    class LVProtocol : public BaseProtocol
    {
    public:
        //  |--LEN--|--VALUE--|
        //  |--LEN--|--mtype--|--idlen--|--id--|--dody--|
        using ptr = std::shared_ptr<LVProtocol>;

        // 检查缓冲区是否可以处理数据
        virtual bool canProcessed(const BaseBuffer::ptr &buf) override
        {
            if (buf->readableSize() < _lenFields_len)
            {
                return false;
            }
            int32_t total_len = buf->peeKInt32();
            if (buf->readableSize() < (total_len + _lenFields_len))
                return false;
            return true;
        }
        // 对接收的消息进行处理并返回消息
        virtual bool onMessage(const BaseBuffer::ptr &buf, BaseMessage::ptr &msg) override
        {
            // 默认缓冲区有一条完整数据
            int32_t total_len = buf->readInt32();
            MType mtype = (MType)buf->readInt32();
            int32_t idlen = buf->readInt32();
            int32_t bodylen = total_len - _mtypeFields_len - _idFields_len - idlen;
            std::string id = buf->retrieveAsString(idlen);
            std::string body = buf->retrieveAsString(bodylen);
            msg = MessageFactory::create(mtype);
            if (msg == nullptr)
            {
                ELOG("消息类型错误，构造消息对象失败")
                return false;
            }
            if (!msg->unserialize(body))
            {
                ELOG("消息反序列化失败");
                return false;
            }
            msg->setId(id);
            msg->setMType(mtype);
            return true;
        }
        // 对消息进行序列化
        virtual std::string serialize(const BaseMessage::ptr &msg) override
        {
            std::string body = msg->serialize();
            std::string id = msg->id();
            auto mtype = htonl((int32_t)msg->mtype());
            int32_t idlen = htonl(id.size());
            int32_t h_total_len = _mtypeFields_len + _idFields_len + id.size() + body.size();
            int32_t n_total_len = htonl(h_total_len);
            std::string result;
            result.reserve(h_total_len);
            result.append((char *)&n_total_len, _lenFields_len);
            result.append((char *)&mtype, _mtypeFields_len);
            result.append((char *)&idlen, _idFields_len);
            result.append(id);
            result.append(body);
            return result;
        }

    private:
        const size_t _lenFields_len = 4;
        const size_t _mtypeFields_len = 4;
        const size_t _idFields_len = 4;
    };

    class ProtocolFactory
    {
    public:
        template <typename... Args>
        static BaseProtocol::ptr create(Args &&...args)
        {
            return std::make_shared<LVProtocol>(std::forward<Args>(args)...);
        }
    };

    class MuduoConnection : public BaseConnection
    {
    public:
        using ptr = std::shared_ptr<MuduoConnection>;
        MuduoConnection(const muduo::net::TcpConnectionPtr &conn, const BaseProtocol::ptr &protocol)
            : _protocol(protocol),
              _conn(conn)
        {
        }
        // 发送数据
        virtual void send(const BaseMessage::ptr &msg) override
        {
            std::string send_body = _protocol->serialize(msg);
            _conn->send(send_body);
        }
        // 关闭连接
        virtual void shutdown() override
        {
            _conn->shutdown();
        }
        // 判断连接是否正常
        virtual bool connected() override
        {
            return _conn->connected();
        }

    private:
        BaseProtocol::ptr _protocol;
        muduo::net::TcpConnectionPtr _conn;
    };

    class ConnectionFactory
    {
    public:
        template <typename... Args>
        static BaseConnection::ptr create(Args &&...args)
        {
            return std::make_shared<MuduoConnection>(std::forward<Args>(args)...);
        }
    };

    class MuduoServer : public BaseServer
    {
    public:
        using ptr = std::shared_ptr<MuduoServer>;
        MuduoServer(int port)
            : _server(&_baseloop, muduo::net::InetAddress("0.0.0.0", port), "MuduoServer", muduo::net::TcpServer::kReusePort),
              _protocol(ProtocolFactory::create())
        {
        }
        // 服务器启动
        virtual void start() override
        {
            _server.setConnectionCallback(std::bind(&MuduoServer::onConnection, this, std::placeholders::_1));
            _server.setMessageCallback(std::bind(&MuduoServer::onMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

            _server.start();
            _baseloop.loop();
        }

    private:
        void onConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            if (conn->connected())
            {
                ILOG("连接成功");
                auto muduo_conn = ConnectionFactory::create(conn, _protocol);
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    _conns.insert(std::make_pair(conn, muduo_conn));
                }
                if (_cb_connection)
                    _cb_connection(muduo_conn);
            }
            else
            {
                ELOG("连接错误或断开");
                BaseConnection::ptr muduo_conn;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _conns.find(conn);
                    if (it == _conns.end())
                        return;
                    muduo_conn = it->second;
                    _conns.erase(conn);
                }
                if (_cb_close)
                    _cb_close(muduo_conn);
            }
        }
        void onMessage(const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buffer, muduo::Timestamp)
        {
            DLOG("有新消息，进行处理");
            DLOG("消息------%ld", buffer->readableBytes());

            auto basebuf = BufferFactory::create(buffer);
            while (1)
            {
                if (_protocol->canProcessed(basebuf) == false)
                {
                    if (basebuf->readableSize() > maxDataSize)
                    {
                        conn->shutdown();
                        ELOG("缓冲区数据过大!");
                        return;
                    }
                    DLOG("数据不可处理");
                    break;
                }
                DLOG("数据可处理");
                BaseMessage::ptr msg;
                bool ret = _protocol->onMessage(basebuf, msg);
                if (ret == false)
                {
                    conn->shutdown();
                    ELOG("缓冲区数据错误!");
                    return;
                }
                DLOG("反序列化成功");
                BaseConnection::ptr base_conn;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _conns.find(conn);
                    if (it == _conns.end())
                    {
                        conn->shutdown();
                        return;
                    }
                    base_conn = it->second;
                }
                DLOG("处理回调");
                if (_cb_message)
                    _cb_message(base_conn, msg);
            }
        }

    private:
        const size_t maxDataSize = (1 << 16);
        BaseProtocol::ptr _protocol;
        muduo::net::EventLoop _baseloop;
        muduo::net::TcpServer _server;
        std::mutex _mutex;
        std::unordered_map<muduo::net::TcpConnectionPtr, BaseConnection::ptr> _conns;
    };

    class ServerFactory
    {
    public:
        template <typename... Args>
        static BaseServer::ptr create(Args &&...args)
        {
            return std::make_shared<MuduoServer>(std::forward<Args>(args)...);
        }
    };

    class MuduoClient : public BaseClient
    {
    public:
        using ptr = std::shared_ptr<MuduoClient>;
        MuduoClient(const std::string &sip, int sport)
            : _protocol(ProtocolFactory::create()),
              _baseloop(_loopthread.startLoop()),
              _downlatch(1),
              _client(_baseloop, muduo::net::InetAddress(sip, sport), "MuduoClient")
        {
        }
        // 建立连接
        virtual void connect() override
        {
            // 设置连接事件回调
            _client.setConnectionCallback(std::bind(&MuduoClient::onConnection, this, std::placeholders::_1));
            // 设置连接消息回调
            _client.setMessageCallback(std::bind(&MuduoClient::onMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
            // 连接服务器
            _client.connect();

            _downlatch.wait();
        }
        // 关闭连接
        virtual void shutdown() override
        {
            _client.disconnect();
        }
        // 发送消息
        virtual bool send(const BaseMessage::ptr &msg) override
        {
            if (connected() == false)
            {
                ELOG("连接已断开")
                return false;
            }
            _conn->send(msg);
            return true;
        }
        // 获取连接
        virtual BaseConnection::ptr connection() override
        {
            return _conn;
        }
        // 判断连接是否正常
        virtual bool connected() override
        {
            return _conn && _conn->connected();
        }

    private:
        void onConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            if (conn->connected())
            {
                ILOG("连接建立！")
                _downlatch.countDown(); // 计数-- 为0时唤醒阻塞
                _conn = ConnectionFactory::create(conn, _protocol);
            }
            else
            {
                ELOG("连接错误！")
                _conn.reset();
            }
        }
        void onMessage(const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buffer, muduo::Timestamp)
        {
            auto basebuf = BufferFactory::create(buffer);
            while (1)
            {
                if (_protocol->canProcessed(basebuf) == false)
                {
                    if (basebuf->readableSize() > maxDataSize)
                    {
                        conn->shutdown();
                        ELOG("缓冲区数据过大!");
                        return;
                    }
                    break;
                }
                BaseMessage::ptr msg;
                bool ret = _protocol->onMessage(basebuf, msg);
                if (ret == false)
                {
                    conn->shutdown();
                    ELOG("缓冲区数据错误!");
                    return;
                }
                if (_cb_message)
                    _cb_message(_conn, msg);
            }
        }

    private:
        const size_t maxDataSize = (1 << 16);
        BaseProtocol::ptr _protocol;
        BaseConnection::ptr _conn;
        muduo::CountDownLatch _downlatch;
        muduo::net::EventLoopThread _loopthread;
        muduo::net::EventLoop *_baseloop;
        muduo::net::TcpClient _client;
    };

    class ClientFactory
    {
    public:
        template <typename... Args>
        static BaseClient::ptr create(Args &&...args)
        {
            return std::make_shared<MuduoClient>(std::forward<Args>(args)...);
        }
    };
}